package com.navinfo.platform.thousandkilo.service;

import com.navinfo.platform.thousandkilo.dto.ThousandKiloData;
import com.navinfo.platform.thousandkilo.pojo.ThousandKiloPojo;
import com.navinfo.platform.thousandkilo.utils.BigDataUtils;
import com.navinfo.platform.thousandkilo.utils.HbaseConverter;
import com.navinfo.platform.thousandkilo.utils.ThousandKiloDataUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.util.HashMap;
import java.util.Map;

public class ThousandKiloService {
    static BigDataUtils buildBigDataUtils(Map<String, String> configMap) {
        BigDataUtils utilService = new BigDataUtils();
        utilService.setTableName(configMap.get("save.thousandKilo.hbase.namespace") + ":"
                + configMap.get("save.thousandKilo.hbase.tableName"));
        utilService.setQuorum(configMap.get("hbase.zookeeper.quorum"));
        utilService.setPort(configMap.get("hbase.zookeeper.property.clientPort"));
        utilService.setParent(configMap.get("zookeeper.znode.parent"));
        return utilService;
    }

    @Deprecated
    static ThousandKiloData queryHbase(BigDataUtils service, Long tid, Integer section) {
        return service.queryHbase(HbaseConverter.getRowKey(tid, section));
    }

    static Map<String, ThousandKiloData> newSaveOrUpdateMap(Map<Long, ThousandKiloData> sectionMap) {
        Map<String, ThousandKiloData> saveOrUpdateMap = new HashMap<>();
        sectionMap.values().forEach(p -> saveOrUpdateMap.put(key(p), p));
        return saveOrUpdateMap;
    }

    static Map<String, ThousandKiloData> initSaveOrUpdateMap(ThousandKiloData p) {
        Map<String, ThousandKiloData> saveOrUpdateMap = new HashMap<>();
        if (p != null) {
            saveOrUpdateMap.put(key(p), p);
        }
        return saveOrUpdateMap;
    }

    static void updateFuel(Map<String, ThousandKiloData> saveOrUpdateMap, ThousandKiloData data, ThousandKiloPojo updateDt) {
        data.setSectionEndFuel(updateDt.getFuel());
        data.setSectionFuel(updateDt.getFuel() - data.getSectionStartFuel());
        data.setSectionEndMileage(updateDt.getMileage());
        data.setSectionMileage(updateDt.getMileage() - data.getSectionStartMileage());
        data.setSectionUrea(data.getSectionUrea() + updateDt.getUrea());
        saveOrUpdateMap.put(key(data), copyOf(data));
    }

    static void createData(Map<String, ThousandKiloData> saveOrUpdateMap, Map<Long, ThousandKiloData> sectionMap,
                           Long tid, Integer section, ThousandKiloPojo updateDt) {
        ThousandKiloData newSectionData = new ThousandKiloData();
        newSectionData.setUid(tid);
        newSectionData.setTid(tid);
        newSectionData.setMileageSection(section);
        newSectionData.setSectionStartFuel(updateDt.getFuel());
        newSectionData.setSectionEndFuel(updateDt.getFuel());
        newSectionData.setSectionFuel(0);
        newSectionData.setSectionStartMileage(updateDt.getMileage());
        newSectionData.setSectionEndMileage(updateDt.getMileage());
        newSectionData.setSectionMileage(0);
        newSectionData.setSectionUrea(0);
        sectionMap.put(tid, newSectionData);
        saveOrUpdateMap.put(key(newSectionData), copyOf(newSectionData));
    }

    static Map<Long, ThousandKiloData> getSectionMap(Dataset<Row> maxSectionByTmn) {
        Map<Long, ThousandKiloData> result = new HashMap<>();
        maxSectionByTmn.foreach(row -> {
            ThousandKiloData data = ThousandKiloDataUtil.parseRowToThousandKiloData(row);
            result.put(data.getTid(), data);
        });
        return result;
    }

    static ThousandKiloData findSection(Map<Long, ThousandKiloData> map, Long tid, Integer section) {
        ThousandKiloData data = map.get(tid);
        if (data != null && data.getMileageSection().equals(section)) {
            return data;
        }
        return null;
    }

    private static String key(ThousandKiloData data) {
        return data.getTid() + "_" + data.getMileageSection();
    }

    private static ThousandKiloData copyOf(ThousandKiloData data) {
        return new ThousandKiloData(data.getUid(), data.getTid(), data.getMileageSection(),
                data.getSectionStartFuel(), data.getSectionEndFuel(), data.getSectionFuel(),
                data.getSectionStartMileage(), data.getSectionEndMileage(), data.getSectionMileage(),
                data.getSectionUrea());
    }
}
